package com.to8to.kitt;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.utils.ThreadUtils;
import org.apache.thrift.ProcessFunction;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.TBase;
import org.apache.thrift.TBaseProcessor;
import org.apache.thrift.protocol.TMessage;
import org.apache.thrift.protocol.TMessageMeta;
import org.apache.thrift.protocol.TMessageType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.to8to.kitt.utils.LogPatterns;
import com.to8to.kitt.utils.LogUtils;
import com.to8to.kitt.utils.NettyUtils;

@Sharable
public class SyncBizChannelHandler extends ChannelHandlerAdapter {
	
	public static Logger logger = LoggerFactory.getLogger(SyncBizChannelHandler.class);  
	
	private int maxTasks = Constants.MAX_BIZ_TASKS;
	
	private int maxBizThreads = Constants.MAX_BIZ_THREADS;
	
	private Map<String, TBaseProcessor<?> > processorMap = new HashMap<String,  TBaseProcessor<?>> ();
	
	private ExecutorService executor;
	
	public SyncBizChannelHandler(TBaseProcessor<?>... processors)
	{
		for(TBaseProcessor<?> p : processors)
		{
			String service = p.getClass().getName();
			
			service = StringUtils.substringBefore(service, "$");
			
			processorMap.put(service, p);
		}
		
		initThreadPool();
	}
	
	public SyncBizChannelHandler(List<TBaseProcessor<?>> processors)
	{
		for(TBaseProcessor<?> p : processors)
		{
			String service = p.getClass().getName();
			
			service = StringUtils.substringBefore(service, "$");
			
			processorMap.put(service, p);
		}
		
		initThreadPool();
	}
	
	public void initThreadPool()
	{
		executor = new ThreadPoolExecutor(maxBizThreads, maxBizThreads,
	            0L, TimeUnit.MILLISECONDS,
	            new LinkedBlockingQueue<Runnable>(maxTasks)
	            , ThreadUtils.newThreadFactory("server-biz-threadpool"));
	}
	
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }


	public int getMaxTasks() {
		return maxTasks;
	}

	public void setMaxTasks(int maxTasks) {
		this.maxTasks = maxTasks;
	}


	public int getMaxBizThreads() {
		return maxBizThreads;
	}

	public void setMaxBizThreads(int maxBizThreads) {
		this.maxBizThreads = maxBizThreads;
	}

	@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) 
    {
    	ThriftMessage tmsg = (ThriftMessage)msg;
		
    	TMessageMeta meta = tmsg.getMeta();
    	
    	LogUtils.openTag(meta.tag);
		
    	logger.debug("===> 收到客户端请求! 对方地址:{} 消息体:{}", ctx.channel().remoteAddress(), tmsg.toString());
    	
    	try
    	{
        	executor.execute(new Runnable(){
				@Override
				public void run() {
					try {
						handle(ctx, tmsg);
			        } 
			    	catch(Throwable e)
			    	{
			    		NettyUtils.exitIfOutOfMemory(e);
			    		
			    		logger.error(LogUtils.logStackTrace(e));
			    	}
				}});
    	}
    	catch(Throwable th)
    	{
    		NettyUtils.exitIfOutOfMemory(th);
    		
    		String errorTxt = "server busy, exceed max queue size: " + maxTasks + "\n" + LogUtils.logStackTrace(th);
    		logger.error(errorTxt);
    		ThriftMessage outMsg = ThriftMessage.newReplyMessage(tmsg.getHead());
    		TMessage h = new TMessage(tmsg.getHead().name, TMessageType.EXCEPTION, tmsg.getSeqid());
			outMsg.setHead(h);
			outMsg.setMeta(meta);
			outMsg.setException(new TApplicationException(TApplicationException.INTERNAL_ERROR, errorTxt));
			
			logger.trace("<==== 返回消息:{}", outMsg.toString());
			
			ToclientChannelFutureListener l = new ToclientChannelFutureListener();
			
			l.start = System.currentTimeMillis();
			
			l.outMsg = outMsg;
			
			ChannelFuture cf = ctx.writeAndFlush(outMsg);
			
			cf.addListener(l);
    	}

    	LogUtils.removeTag();
    }
	
	


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    	
    	NettyUtils.exitIfOutOfMemory(cause);
    	
    	logger.trace(
    			"\n####################################################################################\n"
    			+ "##############  exceptionCaught: \n" + LogUtils.logStackTrace(cause) + "\n" 
    			+ "##############  Channel: \n" + ctx.channel() + "\n"
    			+ "####################################################################################"
    			);
    	
    	
    }
    
    
    public static class ToclientChannelFutureListener implements ChannelFutureListener{
    	
    	public ThriftMessage outMsg;
    	
    	public long start;
    	
		@Override
		public void operationComplete(ChannelFuture future)
				throws Exception {
			
			TMessageMeta meta = outMsg.getMeta();
			
			LogUtils.openTag(meta.tag);
			
			long t = System.currentTimeMillis();
			if(future.isSuccess())
			{
				logger.debug("<==== 回复客户端成功! 对方地址:{} 消息头:{} 发送数据包耗时:{}", future.channel().remoteAddress(), outMsg.getHead(), (t-start));
				logger.trace("<====== 回复包体:{}", outMsg.getBody() != null ? outMsg.getBody() : outMsg.getException());
			}
			else 
				logger.debug("<==== 回复客户端失败!!!!! 对方地址:{} 消息头:{} 发送数据包耗时:{}", future.channel().remoteAddress(), outMsg.getHead(), (t-start));
	
			LogUtils.removeTag();
			
			future.removeListener(this);
		}
	}
    
	private void handle(ChannelHandlerContext ctx, ThriftMessage inMsg) 
	{	
		TMessageMeta meta = inMsg.getMeta();
		
		LogUtils.openTag(meta.tag);
		
		logger.debug("====> 开始业务处理, handle(), 请求消息头:{}", inMsg.toString());
		
		int errorType = 0;
		String errorMsg = null;
		
		ThriftMessage outMsg = ThriftMessage.newReplyMessage(inMsg.getHead());
		outMsg.setMeta(meta);
		
		long t1 = System.currentTimeMillis();
		
		if(inMsg.isCall()) 
    	{
    		String method = meta.method;
    		
    		String service = meta.service;

    		TBaseProcessor<?> p = null;
    		
    		if(!StringUtils.isEmpty(service))
    			p = processorMap.get(service);
    		
    		if(p == null && processorMap.size() == 1)
    		{
    			p = processorMap.values().iterator().next();
    			logger.warn("找不到服务处理函数: {}, 取一个默认的: {}", service, p.getClass());
    		}
    		
    		if(p != null)
    		{
    			ProcessFunction fn = (ProcessFunction) p.getProcessMapView().get(method);
    			if(fn != null)
    			{
    				try {
    					
    					Object iface = p.getIface();
    					
    					TBase<?,?> msgBody = inMsg.getBody();
    					
    					if(iface != null && msgBody != null)
    					{
        					TBase<?,?> result = fn.getResult(iface, msgBody);
        					outMsg.setBody(result);
    					}
    					else 
    					{
    						errorMsg = LogPatterns.FAIL_TAG + " iface为空，或者消息体为空, iface:" + iface +", body: " + msgBody;
    						errorType = TApplicationException.INVALID_PROTOCOL;
    					}
    				} catch (Throwable e) {
    					
    					NettyUtils.exitIfOutOfMemory(e);
    					
    					errorMsg = LogPatterns.FAIL_TAG + " 业务处理函数异常: 消息体:" + inMsg.getBody() +"\n异常信息:" + LogUtils.logStackTrace(e);
    					errorType = TApplicationException.INTERNAL_ERROR;
    				}
    			}
    			else 
    			{
    				errorMsg = LogPatterns.FAIL_TAG + " ProcessFunction not found:" + method;
    				errorType = TApplicationException.INVALID_PROTOCOL;
    			}
    		}
    		else 
    		{
    			errorMsg = LogPatterns.FAIL_TAG + " TBaseProcessor not found:" + service;
    			errorType = TApplicationException.INVALID_PROTOCOL;
    		}
    	}
		else
		{
			errorMsg = LogPatterns.FAIL_TAG + " 消息类型出错，不应该收到此类型的消息: " + inMsg.getHead();
			errorType = TApplicationException.INVALID_MESSAGE_TYPE;
		}
		
		long t2 = System.currentTimeMillis();
		
		String result = "成功";
		if(errorMsg != null)
		{
			TMessage h = new TMessage(inMsg.getHead().name, TMessageType.EXCEPTION, inMsg.getSeqid());
			outMsg.setHead(h);
			outMsg.setException(new TApplicationException(errorType, errorMsg));
			logger.error(errorMsg);
			result = "失败";
		}
		
		long delta = t2-t1;
		
		if(delta > 200)
			logger.warn("完成业务处理, 处理结果:{}, 处理业务耗时:{} ms, 超时了!!!, 回复消息头:{}", result, delta, outMsg.getHead());
		else
			logger.debug("完成业务处理, 处理结果:{}, 处理业务耗时:{} ms, 回复消息头:{}", result, delta, outMsg.getHead());
		
		logger.trace("<==== 返回消息:{}", outMsg.toString());
		
		ToclientChannelFutureListener l = new ToclientChannelFutureListener();
		
		l.start = t2;
		
		l.outMsg = outMsg;
		
		ChannelFuture cf = ctx.writeAndFlush(outMsg);
//		ChannelFuture cf = ctx.write(outMsg);
		
		cf.addListener(l);
		
		LogUtils.removeTag();
	}
}
